package com.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 模拟消费逻辑
 * 监听一个主题,每当这个主题中出现一条消息,消费逻辑就执行一次
 */
public class MyConsumer04 {
    @Test
    public void consume()throws Exception{
    //1. 创建消费对象consumer
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer();
        //连接namesrv
        consumer.setNamesrvAddr("localhost:9876");
        //创建一个分组
        consumer.setConsumerGroup("test-consumer-01");
        //绑定监听的主题
        consumer.subscribe("test-topic-01","*");
    //2. 创建消费者监听器,获取消息,执行消费逻辑
        //new一个匿名内部类 监听器
        MessageListenerConcurrently messageListener=
                new MessageListenerConcurrently() {
                    /**
                     * 监听器每当主题出现一条消息,
                     * 就会调用一次consumeMessage
                     * list: 每次消费的消息对象 虽然用list存储,但是每个list就只有一个元素
                     * return: 消费逻辑的返回状态
                     * 1. CONSUME_SUCCESS 消费成功
                     * 2. RECONSUME_LATER 等待一会在重新消费(本次消费不顺利) 会导致消息重新发送 retry机制
                     */
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        //从消息对象中解析 消息体byte[],打印字符串
                        MessageExt messageExt = list.get(0);
                        byte[] body = messageExt.getBody();
                        String message=new String(body, StandardCharsets.UTF_8);
                        System.out.println("当前消费者04,接收到消息:"+message);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                };
        //监听器交给消费者
        consumer.setMessageListener(messageListener);
    //3. 开启连接,开始监听. 保证测试代码线程不死
        consumer.start();
        while(true);
    }
}
